#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include "cJSON.h"
#include "MQTTClient.h"
#include "shmem.h"
#include "msg_queue_peer.h"
#include <sqlite3.h>
#include <time.h>

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define QOS 1
#define TIMEOUT 10000L

struct shm_param para;    // 共享内存
struct dapeng_date *shm;  // 共享内存映射
struct dapengkongzhi msg; // 消息队列结构体
FILE *fp;                 // 网关端的node.json
cJSON *root;              // 序列化的node.json
int flag = 0;             // 刷新上报标志位
sqlite3 *db;
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;
int ch;
void history();
void showHistory();
void showTimeHistory(char *times);

volatile MQTTClient_deliveryToken deliveredtoken;
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message) // 接收qt发送的消息
{
    fp = fopen("./node.json", "r+");
    if (fp == NULL)
    {
        perror("fopen file err");
        return -1;
    }
    fseek(fp, 0, SEEK_END); // 获取文件大小
    long l = ftell(fp);
    rewind(fp);                         // 将文件指针重置到文件开头
    char *node = (char *)malloc(l + 1); // 分配内存存放文件内容
    if (node == NULL)
    {
        perror("Memory allocation error");
        fclose(fp);
        return -1;
    }
    if (fread(node, 1, l, fp) != l) // 读取文件内容到node数组
    {
        perror("File read error");
        free(node);
        fclose(fp);
        return -1;
    }
    root = cJSON_Parse(node);
    cJSON *old_report = cJSON_GetObjectItem(root, "report"); // 原本的report节点
    cJSON *roots = cJSON_Parse(message->payload);            // 读取客户端发送过来的json
    cJSON *type = cJSON_GetObjectItem(roots, "type");        // 读取type节点
    if (type->valueint == 3)                                 // 判断节点类型为模式调节
    {
        cJSON *data = cJSON_GetObjectItem(roots, "data");    // 读取data节点
        cJSON *type = cJSON_GetObjectItem(data, "type");     // 读取新的模式type
        cJSON *period = cJSON_GetObjectItem(data, "period"); // 读取新的模式period
        cJSON *change = cJSON_CreateObject();                // 创建一个新的节点
        cJSON_AddItemToObject(change, "type", cJSON_CreateNumber(type->valueint));
        cJSON_AddItemToObject(change, "period", cJSON_CreateNumber(period->valueint));
        cJSON_ReplaceItemInObject(root, "report", change);
        char *updatedroot = cJSON_Print(root); // 将修改后的 JSON 写回文件
        freopen("./node.json", "w", fp);
        fwrite(updatedroot, 1, strlen(updatedroot), fp);
        fflush(fp); // 刷新缓冲区确保内容写入磁盘
        free(updatedroot);
    }
    if (type->valueint == 2) // 判断节点类型为控制模式
    {
        cJSON *data = cJSON_GetObjectItem(roots, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *val = cJSON_GetObjectItem(data, "val");

        msg.type = 2;
        printf("***********************\n");
        msg.kongzhi.key = key->valueint;
        strcpy(msg.kongzhi.value, val->valuestring);
        msg_queue_send("daxianmuxiaoxi", &msg, sizeof(msg) - sizeof(long), 0);
        printf("%d\n", msg.kongzhi.key);
        printf("%s\n", msg.kongzhi.value);
    }
    if (type->valueint == 1) // 判断节点类型为采集模式
    {
        flag = 1;
    }
    if (type->valueint == 4) // 判断节点类型为历史记录查询信号
    {
        
        cJSON *time = cJSON_GetObjectItem(roots, "time"); // 读取time节点
        if (strcmp(time->valuestring, "all")==0)
        {
            showHistory();
        }
        else
        {
            showTimeHistory(time->valuestring);
        }
    }

    cJSON_Delete(root);
    cJSON_Delete(roots);
    free(node);
    fclose(fp);

    // 直接调用 shell 命令来重启程序
    int result = system("./shujushangbao");
    if (result == -1)
    {
        perror("system() failed to restart the process");
        return -1;
    }

    MQTTClient_freeMessage(&message);
    return 0;
}
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
int main(int argc, char *argv[])
{
    // 创建或打开数据库
    if (sqlite3_open("./history.db", &db)!= 0)
    {
        printf("打开或创建数据库失败:%s", sqlite3_errmsg(db));
        return -1;
    }
    printf("打开数据库成功\n");

    char *errmsg = NULL;
    // 创建历史记录表(时间,温度,湿度,光照强度,二氧化碳浓度,氧气浓度,土壤湿度)
    if(sqlite3_exec(db, "create table lishijilu(time char,wendu double,shidu double,guangzhao int,eryanghuatan double,yangqi double,turang double);", NULL, NULL, &errmsg))
    {
        printf("创建用户表失败:%s\n", errmsg);
    }

    shm_init(&para, "daxaingmugongxiang", sizeof(struct dapeng_date));
    shm = shm_getaddr(&para);

    while (1)
    {
        fp = fopen("./node.json", "r+");
        if (fp == NULL)
        {
            perror("fopen file err");
            sleep(5);
        }
        else
        {
            break;
        }
    }
    fseek(fp, 0, SEEK_END); // 获取文件大小
    long l = ftell(fp);
    rewind(fp);                         // 将文件指针重置到文件开头
    char *node = (char *)malloc(l + 1); // 分配内存存放文件内容
    if (node == NULL)
    {
        perror("Memory allocation error");
        fclose(fp);
        return -1;
    }
    if (fread(node, 1, l, fp) != l) // 读取文件内容到node数组
    {
        perror("File read error");
        free(node);
        fclose(fp);
        return -1;
    }
    node[l] = '\0'; // 确保字符串以NULL结尾
    fclose(fp);     // 读取完文件后关闭文件
    // 解析node.json数据
    root = cJSON_Parse(node);
    cJSON *version = cJSON_GetObjectItem(root, "version");
    strcpy(shm->vs, version->valuestring);

    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON *type = cJSON_GetObjectItem(report, "type");
    cJSON *period = cJSON_GetObjectItem(report, "period");
    shm->lx.type = type->valueint;
    shm->lx.period = period->valueint;

    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    cJSON *mqttaddr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *mqttport = cJSON_GetObjectItem(mqtt_server, "port");
    strcpy(shm->ms.addr, mqttaddr->valuestring);
    shm->ms.port = mqttport->valueint;

    cJSON *mb_dev = cJSON_GetObjectItem(root, "mb_dev");
    cJSON *mdaddr = cJSON_GetObjectItem(mb_dev, "addr");
    cJSON *mdport = cJSON_GetObjectItem(mb_dev, "port");
    strcpy(shm->md.addr, mdaddr->valuestring);
    shm->md.port = mdport->valueint;

    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *stm32data = cJSON_GetObjectItem(stm32, "data");
    int n = cJSON_GetArraySize(stm32data);
    shm->stm32_count = n;
    for (int i = 0; i < n; i++)
    {
        cJSON *item = cJSON_GetArrayItem(stm32data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *name = cJSON_GetObjectItem(item, "name");
        cJSON *type = cJSON_GetObjectItem(item, "type");
        shm->st[i].key = key->valueint;
        strcpy(shm->st[i].name, name->valuestring);
        shm->st[i].type = type->valueint;
    }

    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    cJSON *modbusdata = cJSON_GetObjectItem(modbus, "data");
    int n1 = cJSON_GetArraySize(modbusdata);
    shm->modbus_count = n1;
    for (int j = 0; j < n1; j++)
    {
        cJSON *item = cJSON_GetArrayItem(modbusdata, j);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *name = cJSON_GetObjectItem(item, "name");
        cJSON *addr = cJSON_GetObjectItem(item, "addr");
        cJSON *type = cJSON_GetObjectItem(item, "type");
        shm->mb[j].key = key->valueint;
        strcpy(shm->mb[j].name, name->valuestring);
        shm->mb[j].addr = addr->valueint;
        shm->mb[j].type = type->valueint;
    }
    // shm_write(&para, shm, sizeof(*shm));
    cJSON_Delete(root);

    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, "/app/data/down", QOS);
    char *old_data = shujucaiji();
    while (1)
    {
        if (shm->lx.type == 0) // 刷新上报
        {
            if (flag == 1)
            {
                char *p = shujucaiji();
                pubmsg.payload = p;
                pubmsg.payloadlen = (int)strlen(p);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, "/app/data/up", &pubmsg, &token);
                flag = 0;
                history();
                printf("刷新上报成功\n");
            }
        }
        else if (shm->lx.type == 1) // 定时上报
        {
            char *p = shujucaiji();
            pubmsg.payload = p;
            pubmsg.payloadlen = (int)strlen(p);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, "/app/data/up", &pubmsg, &token);
            printf("定时上报成功\n");
            history();
            sleep(shm->lx.period);
        }
        else if (shm->lx.type == 2) // 变化上报
        {
            sleep(1);
            char *new_data = shujucaiji();
            if (old_data != NULL && new_data != NULL && strcmp(old_data, new_data) != 0)
            {
                pubmsg.payload = new_data;
                pubmsg.payloadlen = (int)strlen(new_data);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, "/app/data/up", &pubmsg, &token);
                history();
                printf("变化上报成功\n");
            }
            if (old_data != new_data)
            {
                if (old_data != NULL)
                {
                    free(old_data);
                }
                old_data = new_data;
            }
        }
    }
    MQTTClient_unsubscribe(client, "/app/data/down");
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}
char *shujucaiji()
{
    // JSON序列化
    cJSON *root = cJSON_CreateObject();
    cJSON *stm = cJSON_CreateObject();
    cJSON *modbus = cJSON_CreateObject();
    cJSON *stdata = cJSON_CreateArray();
    cJSON *mbdata = cJSON_CreateArray();
    for (int i = 0; i < shm->stm32_count; i++)
    {
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(shm->st[i].key));
        cJSON_AddItemToObject(item, "name", cJSON_CreateString(shm->st[i].name));
        cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(shm->st[i].type));
        if (shm->st[i].type == 1)
        {
            cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm->st[i].value.val_bool));
        }
        else if (shm->st[i].type == 2)
        {
            cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm->st[i].value.val_int));
        }
        else if (shm->st[i].type == 3)
        {
            cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm->st[i].value.val_double));
        }
        cJSON_AddItemToArray(stdata, item);
    }
    for (int j = 0; j < shm->modbus_count; j++)
    {
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(shm->mb[j].key));
        cJSON_AddItemToObject(item, "name", cJSON_CreateString(shm->mb[j].name));
        cJSON_AddItemToObject(item, "addr", cJSON_CreateNumber(shm->mb[j].addr));
        cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(shm->mb[j].type));
        if (shm->mb[j].type == 1)
        {
            cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm->mb[j].value.val_bool));
        }
        if (shm->mb[j].type == 2)
        {
            cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm->mb[j].value.val_int));
        }
        if (shm->mb[j].type == 3)
        {
            cJSON_AddItemToObject(item, "val", cJSON_CreateNumber(shm->mb[j].value.val_double));
        }
        cJSON_AddItemToArray(mbdata, item);
    }
    cJSON_AddItemToObject(stm, "data", stdata);
    cJSON_AddItemToObject(modbus, "data", mbdata);
    cJSON_AddItemToObject(root, "stm32", stm);
    cJSON_AddItemToObject(root, "modbus", modbus);
    char *p = cJSON_Print(root);
    return p;
}
void history()
{
    char sql1[1024];
    char *errmsg = NULL;
    char **result = NULL;
    int rows, columns;
    time_t t;
    struct tm *tm;
    char stime[100];
    time(&t);
    tm = localtime(&t);
    sprintf(stime, "%d-%d-%d %d-%d-%d", tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday, tm->tm_hour, tm->tm_min, tm->tm_sec);
    snprintf(sql1, sizeof(sql1), "insert into lishijilu values ('%s',%lf,%lf,%d,%lf,%lf,%lf);", stime, shm->st[0].value.val_double, shm->st[1].value.val_double, shm->st[2].value.val_int, shm->mb[0].value.val_double, shm->mb[1].value.val_double, shm->mb[2].value.val_double);
    if (sqlite3_exec(db, sql1, NULL, NULL, &errmsg) != 0)
    {
        printf("放入历史记录失败: %s", errmsg);
    }
}
void showHistory()
{
    char buf[10240];
    char sql[1100];
    char *errmsg = NULL;
    char **result = NULL;
    int rows, columns;
    sprintf(sql, "select * from lishijilu;");
    int rc = sqlite3_get_table(db, sql, &result, &rows, &columns, &errmsg);
    if (rc != 0)
    {
        printf("查询历史记录失败: %s", errmsg);
    }
    else
    {
        if (rows > 0)
        {
            for (int i = 1; i <= rows; ++i)
            {
                for (int j = 0; j < columns; ++j)
                {
                    strcat(buf, result[i * columns + j]);
                    strcat(buf, " ");
                } 
                strcat(buf, "\n");
            }
        }
    }
    cJSON *root = cJSON_CreateObject();
    cJSON *history = cJSON_CreateString(buf);
    cJSON_AddItemToObject(root, "history", history);
    char *p = cJSON_Print(root);

    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, "/app/data/up", &pubmsg, &token);
    cJSON_Delete(root);
    free(p);
}
void showTimeHistory(char *times)
{
    char buf[10240];
    char sql[1100];
    char *errmsg = NULL;
    char **result = NULL;
    int rows, columns;
    sprintf(sql, "select * from lishijilu where time like '%%%s%%'", times);
    int rc = sqlite3_get_table(db, sql, &result, &rows, &columns, &errmsg);
    if (rc != 0)
    {
        printf("查询历史记录失败: %s", errmsg);
    }
    else
    {
        if (rows > 0)
        {
            for (int i = 1; i <= rows; ++i)
            {
                for (int j = 0; j < columns; ++j)
                {
                    strcat(buf, result[i * columns + j]);
                    strcat(buf, " ");
                } 
                strcat(buf, "\n");
            }
        }
        else
        {
            strcpy(buf,"当天没有消息记录");
        }
    }
    cJSON *root = cJSON_CreateObject();
    cJSON *history = cJSON_CreateString(buf);
    cJSON_AddItemToObject(root, "history", history);
    char *p = cJSON_Print(root);

    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, "/app/data/up", &pubmsg, &token);
    cJSON_Delete(root);
    free(p);
}
